package com.chenjj.bigdata.hbase.client.impl;


import com.chenjj.bigdata.hbase.client.*;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.FilterList;
import org.apache.hadoop.hbase.filter.PageFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.LinkedHashMap;
import java.util.List;
import java.util.Map;

public class HbaseClientImpl extends AbstractHbaseClient {

	private Logger logger = LoggerFactory.getLogger(getClass());

    public HbaseClientImpl(String zkAdress, String zkPort) {
        super(zkAdress, zkPort);
    }

	public HbaseClientImpl(int poolSize, String zkAdress, String zkPort) {
		super(poolSize, zkAdress, zkPort);
	}


	public List<Map<String, String>> scan(String tableName, String startRowkey,
                                          String stopRowkey, HbaseColumn[] hc, Filter filter)
			throws IOException {
		return scan(tableName, startRowkey, stopRowkey, hc, filter,null);
	}

    @Override
    public List<Map<String, String>> scan(String tableName, String startRowkey, String stopRowkey, HbaseColumn[] hc, Filter filter, HBasePagination page) throws IOException {
        Scan scan = new Scan();
        scan.setFilter(filter);
        scan.setStartRow(Bytes.toBytes(startRowkey));
        scan.setStopRow(Bytes.toBytes(stopRowkey));
        scan.setCaching(500);
        //分页信息
        if (page != null){
            long maxResultSize = page.getPageSize();
            PageFilter pageFilter = new PageFilter(maxResultSize);
            scan.setMaxResultSize(maxResultSize);
            Filter f = scan.getFilter();
            if(f != null && f instanceof FilterList){
                ((FilterList) f).addFilter(pageFilter);
                scan.setFilter(f);
            }else{
                FilterList filterList = new FilterList();
                filterList.addFilter(pageFilter);
                if (f != null){
                    filterList.addFilter(f);
                }
                scan.setFilter(filterList);
            }
        }

        //返回的列信息
        if (hc != null) {
            for (int i = 0; i < hc.length; i++) {
                HbaseColumn c = hc[i];
                scan.addColumn(Bytes.toBytes(c.getFamilyName()),
                        Bytes.toBytes(c.getQualifier()));
            }
        }

        ResultScanner resultScanner = getResultScanner(tableName, scan);

        List<Map<String, String>> result = new ArrayList<Map<String, String>>();
        if (resultScanner != null) {
            for (Result r : resultScanner) {
                Map<String, String> row = new LinkedHashMap<String, String>();
                // 得到单元格集合
                List<Cell> cs = r.listCells();
                for (Cell cell : cs) {
                    // 取到rowkey
                    String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                    // 取到时间戳
                    //long timestamp = cell.getTimestamp();
                    // 取到族列
                    //String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    // 取到修饰名
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    // 取到值
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    row.put("rowkey",rowkey);
                    row.put(qualifier, value);
                }
                result.add(row);
            }
        }
        return result;
    }


    public List<Map<String, String>> scan(String tableName, String startRowkey,
			String stopRowkey) throws IOException {
		return scan(tableName, startRowkey, stopRowkey, null, null);
	}

	public List<Map<String, String>> scan(String tableName, String startRowkey,
			String stopRowkey, Filter filter) throws IOException {
		return scan(tableName, startRowkey, stopRowkey, null, filter);
	}

    public void put(String tableName, List<HbaseRow> rows) {
		Table table = null;
		try {
			table = getConnection().getTable(
					TableName.valueOf(tableName.getBytes()));
			List<Put> listPut = new ArrayList<Put>();
			for (int i = 0; i < rows.size(); i++) {
				Put put = new Put(Bytes.toBytes(rows.get(i).getRowkey()));
				for (int j = 0; j < rows.get(i).getColumn().length; j++) {
					HbaseColumn column = rows.get(i).getColumn()[j];
					put.addColumn(Bytes.toBytes(column.getFamilyName()),
							Bytes.toBytes(column.getQualifier()),
							Bytes.toBytes(column.getValue()));
				}
				listPut.add(put);
			}
			table.put(listPut);
		} catch (Exception e) {
			logger.error("HbaseClient put failed ", e);
			throw new HbaseClientException(e);
		} finally {
			if (table != null) {
				try {
					table.close();
				} catch (IOException e) {
					logger.error("HbaseClient table close failed ", e);
					e.printStackTrace();
				}
			}
		}

	}

	//TODO 考虑 事务回滚的问题？
	public void put(String tableName, List<HbaseRow> rows,
			List<String> indexNames) {
//
//		if (solrClient == null) {
//			throw new HbaseClientException("solrClient is null");
//		}
//
//		put(tableName, rows);
//		Collection<SolrInputDocument> docs = new ArrayList<SolrInputDocument>();
//		SolrInputDocument doc = null;
//		//TODO 复杂度略高，有空考虑换一种实现方式
//		for (int i = 0; i < rows.size(); i++) {
//			doc = new SolrInputDocument();
//			HbaseRow row = rows.get(i);
//			for (int j = 0; j < indexNames.size(); j++) {
//					String indexName = indexNames.get(j);
//				for (int j2 = 0; j2 <row.getColumn().length; j2++) {
//					HbaseColumn column = row.getColumn()[j2];
//					//rowkey
//					if("rowkey".equals(indexName)){
//						doc.addField("rowkey", row.getRowkey());
//						break;
//					}else if(column.getQualifier().equalsIgnoreCase(indexName)){ //index name
//						doc.addField(indexName,column.getValue());
//					}
//				}
//			}
//			if(doc.getField("rowkey") == null  || StringUtils.isEmpty(doc.getField("rowkey").getValue())){
//				throw new RuntimeException("not found 'rowkey' in indexNames");
//			}
//			docs.add(doc);
//			logger.info("Solr document collection added one: {}",doc);
//		}
//		solrClient.add(docs);
//		logger.info("Solr indexs create/update succefully!");
	}

	public List<Map<String, String>> get(String tableName,
			List<String> rowkeys, HbaseColumn[] hc) {
		return get(tableName,rowkeys,hc,null);
	}

    @Override
    public List<Map<String, String>> get(String tableName, List<String> rowkeys, HbaseColumn[] hc, Filter filter) {
        List<Get> gets = new ArrayList<Get>();
        for (int i = 0; i <rowkeys.size(); i++) {
            Get get = new Get(Bytes.toBytes(rowkeys.get(i)));

            if(hc != null){
                for (int j = 0; j < hc.length; j++) {
                    HbaseColumn c = hc[j];
                    get.addColumn(Bytes.toBytes(c.getFamilyName()),Bytes.toBytes(c.getQualifier()));
                }
            }


            if (filter != null){
                get.setFilter(filter);
            }

            gets.add(get);
        }

        Table table = null;
        Result[] results =null;
        try {
            table = getConnection().getTable(TableName.valueOf(tableName.getBytes()));
            results = table.get(gets);
        } catch (IllegalArgumentException | IOException e) {
            throw new RuntimeException(e);
        }


        List<Map<String, String>> result = new ArrayList<Map<String, String>>();
        if (results != null) {
            for (Result r : results) {
                Map<String, String> row = new LinkedHashMap<String, String>();
                // 得到单元格集合
                List<Cell> cs = r.listCells();
                for (Cell cell : cs) {
                    // 取到rowkey
                    String rowkey = Bytes.toString(CellUtil.cloneRow(cell));
                    // 取到时间戳
                    //long timestamp = cell.getTimestamp();
                    // 取到族列
                    //String family = Bytes.toString(CellUtil.cloneFamily(cell));
                    // 取到修饰名
                    String qualifier = Bytes.toString(CellUtil.cloneQualifier(cell));
                    // 取到值
                    String value = Bytes.toString(CellUtil.cloneValue(cell));
                    row.put("rowkey",rowkey);
                    row.put(qualifier, value);
                }
                result.add(row);
            }
        }
        return result;
    }

    @Override
    public Connection getConnection() {
        return super.getConnection();
    }

}
